package com.deep.cdc;

import com.deep.bean.ConnectConfig;
import com.deep.bean.TransformConfig;
import com.deep.common.ArgsConfig;
import com.deep.common.ArgsUtil;
import com.deep.common.ProfileConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * SELECT count(1) as c , actid from scdata.t_wf_actinst GROUP BY actid  HAVING c > 1;
 *
 * SELECT * from scdata.t_wf_actinst where actid = 1347974;
 * SELECT * from scdata.t_wf_actinst where actid = 1347976;
 *
 * SELECT * from scdata.t_wf_actinst where actid = 1347976 and recid = 850229;
 *
 * SELECT count(1) as c , actid, recid from scdata.t_wf_actinst GROUP BY actid, recid  HAVING c > 1;
 *
 * delete from scdata.t_wf_actinst where actid = 1347976 and recid = 850229 and PARTID is null;
 */
@Slf4j
public class SyncActInst {
    public static void main(String[] args) {
        ArgsConfig argsConfig = ArgsUtil.getMap(args);
        Integer recid = argsConfig.getRecid();

        TransformConfig transformConfig = ProfileConfig.getTransformConfig(argsConfig.getProfile());
        ConnectConfig sourceConfig = transformConfig.getSource();
        ConnectConfig targetConfig = transformConfig.getTarget();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //创建用于输出的表
        String createInputDDL = "create table source_t_wf_actinst  " +
                "( " +
                "ACTID  int   , " +
                "PROCCODE  string   , " +
                "PROCID  int   , " +
                "ACTCODE  string   , " +
                "ACTDEFID  int   , " +
                "ACTDEFNAME  string   , " +
                "RECID  int   , " +
                "PARTID  int   , " +
                "PARTNAME  string   , " +
                "UNITID  int   , " +
                "UNITNAME  string   , " +
                "ROLEPARTID  int   , " +
                "ROLEPARTNAME  string   , " +
                "SUPERVISESTATEID  int   , " +
                "SUPERVISESTATENAME  string   , " +
                "ACCREDITSTATEID  int   , " +
                "ACCREDITSTATENAME  string   , " +
                "ACCREDITWORKITEMID  string   , " +
                "CREATETIME  timestamp(0)   , " +
                "DEADLINE  timestamp(0)   , " +
                "WARNINGTIME  timestamp(0)   , " +
                "ENDTIME  timestamp(0)   , " +
                "PREACTID  string   , " +
                "PREPARTID  string   , " +
                "PREPARTNAME  string   , " +
                "PREACTDEFID  string   , " +
                "PREACTDEFNAME  string   , " +
                "PREACTOPINION  string   , " +
                "NEXTACTID  string   , " +
                "NEXTPARTID  string   , " +
                "NEXTPARTNAME  string   , " +
                "NEXTACTDEFID  string   , " +
                "NEXTACTDEFNAME  string   , " +
                "ACTTIMESTATEID  int   , " +
                "ACTLIMITINFO  string   , " +
                "ACTTIMEUNIT  int   , " +
                "ACTLIMIT  decimal   , " +
                "ACTWARNINGLIMIT  decimal   , " +
                "ACTUSEDTIME  decimal   , " +
                "ACTREMAINTIME  decimal   , " +
                "ACTUSEDTIMECHAR  string   , " +
                "ACTREMAINTIMECHAR  string   , " +
                "REFRESHFLAG  int   , " +
                "REFRESHSTARTTIME  timestamp(0)   , " +
                "REFRESHTIME  timestamp(0)   , " +
                "REFRESHSTOPCOUNT  int   , " +
                "TIMEADDUPFLAG  int   , " +
                "BACKACTID  string   , " +
                "REPLYBACKACTID  string   , " +
                "TRANSWORKITEMID  int   , " +
                "TRANSWORKITEMNAME  string   , " +
                "READFLAG  tinyint   , " +
                "READTIME  timestamp(0)   , " +
                "ROLLBACKENABLED  int   , " +
                "TRANSOPINION  string   , " +
                "ARDINSTDETAILNO  int   , " +
                "ACTUSEDTIMECUR  double   , " +
                "ISTIMINGSTOP  int   , " +
                "ACTUSEDTIMEBEFORESTOP  double   , " +
                "ACTRESTART  timestamp   , " +
                "ACTUSEDTIMECURCHAR  string   , " +
                "OVERTIME  tinyint   , " +
                "ACCEPTTIME  timestamp   , " +
                "ACCEPTPARTID  int   , " +
                "ACCEPTPARTNAME  string  , " +
                "  PRIMARY KEY (`ACTID`,`RECID`) NOT ENFORCED  " +
                ")WITH (  " +
                "    'connector' = 'mysql-cdc', " +
                "    'hostname' = '" + sourceConfig.getHostname() + "',  " +
                "    'port' = '" + sourceConfig.getPort() + "',   " +
                "    'username' = '" + sourceConfig.getUsername() + "',  " +
                "    'password' = '" + sourceConfig.getPassword() + "',   " +
                "    'database-name' = '" + sourceConfig.getDatabaseName() + "',  " +
                "    'table-name' = 't_wf_actinst'," +
                "   'debezium.skipped.operations'='d' " +
                ") ";
        log.info("createInputDDL = {}", createInputDDL);
        tableEnv.executeSql(createInputDDL);


        //创建用于输出的表
        String createOutputDDL = "create table target_t_wf_actinst " +
                "( " +
                "ACTID  int   , " +
                "PROCCODE  string   , " +
                "PROCID  int   , " +
                "ACTCODE  string   , " +
                "ACTDEFID  int   , " +
                "ACTDEFNAME  string   , " +
                "RECID  int   , " +
                "PARTID  int   , " +
                "PARTNAME  string   , " +
                "UNITID  int   , " +
                "UNITNAME  string   , " +
                "ROLEPARTID  int   , " +
                "ROLEPARTNAME  string   , " +
                "SUPERVISESTATEID  int   , " +
                "SUPERVISESTATENAME  string   , " +
                "ACCREDITSTATEID  int   , " +
                "ACCREDITSTATENAME  string   , " +
                "ACCREDITWORKITEMID  string   , " +
                "CREATETIME  timestamp(0)   , " +
                "DEADLINE  timestamp(0)   , " +
                "WARNINGTIME  timestamp(0)   , " +
                "ENDTIME  timestamp(0)   , " +
                "PREACTID  string   , " +
                "PREPARTID  string   , " +
                "PREPARTNAME  string   , " +
                "PREACTDEFID  string   , " +
                "PREACTDEFNAME  string   , " +
                "PREACTOPINION  string   , " +
                "NEXTACTID  string   , " +
                "NEXTPARTID  string   , " +
                "NEXTPARTNAME  string   , " +
                "NEXTACTDEFID  string   , " +
                "NEXTACTDEFNAME  string   , " +
                "ACTTIMESTATEID  int   , " +
                "ACTLIMITINFO  string   , " +
                "ACTTIMEUNIT  int   , " +
                "ACTLIMIT  double   , " +
                "ACTWARNINGLIMIT  double   , " +
                "ACTUSEDTIME  double   , " +
                "ACTREMAINTIME  double   , " +
                "ACTUSEDTIMECHAR  string   , " +
                "ACTREMAINTIMECHAR  string   , " +
                "REFRESHFLAG  int   , " +
                "REFRESHSTARTTIME  timestamp(0)   , " +
                "REFRESHTIME  timestamp(0)   , " +
                "REFRESHSTOPCOUNT  int   , " +
                "TIMEADDUPFLAG  int   , " +
                "BACKACTID  string   , " +
                "REPLYBACKACTID  string   , " +
                "TRANSWORKITEMID  int   , " +
                "TRANSWORKITEMNAME  string   , " +
                "READFLAG  tinyint   , " +
                "READTIME  timestamp(0)   , " +
                "ROLLBACKENABLED  int   , " +
                "TRANSOPINION  string   , " +
                "ARDINSTDETAILNO  int   , " +
                "ACTUSEDTIMECUR  double   , " +
                "ISTIMINGSTOP  int   , " +
                "ACTUSEDTIMEBEFORESTOP  double   , " +
                "ACTRESTART  timestamp   , " +
                "ACTUSEDTIMECURCHAR  string   , " +
                "OVERTIME  tinyint   , " +
                "ACCEPTTIME  timestamp   , " +
                "ACCEPTPARTID  int   , " +
                "ACCEPTPARTNAME  string  , " +
                "  PRIMARY KEY (`ACTID`,`RECID`) NOT ENFORCED " +
                ")WITH ( " +
                " 'connector' = 'jdbc', " +
                " 'driver' = 'com.mysql.cj.jdbc.Driver', " +
                " 'url' = 'jdbc:mysql://"+ targetConfig.getHostname() + ":" + targetConfig.getPort() + "/" + targetConfig.getDatabaseName() + "?rewriteBatchedStatements=true&useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=UTC',  " +
                " 'username' = '" + targetConfig.getUsername() + "',  " +
                "  'password' = '" + targetConfig.getPassword() + "',   " +
                " 'table-name' = 't_wf_actinst', " +
                " 'connection.max-retry-timeout' = '60s' " +
                ")";
        log.info("createOutputDDL = {}", createOutputDDL);

        tableEnv.executeSql(createOutputDDL);


        TableResult tableResult = tableEnv.executeSql("insert into target_t_wf_actinst select * from source_t_wf_actinst where RECID >  " + recid);

        tableResult.print();

    }
}
